package com.shujia.flink.core

import org.apache.flink.api.common.functions.RuntimeContext
import org.apache.flink.api.common.state.{ValueState, ValueStateDescriptor}
import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.functions.KeyedProcessFunction
import org.apache.flink.streaming.api.scala._
import org.apache.flink.util.Collector

object Demo7ValueState {
  def main(args: Array[String]): Unit = {
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment


    val linesDS: DataStream[String] = env.socketTextStream("master", 8888)

    val kvDS: DataStream[(String, Int)] = linesDS.flatMap(_.split(",")).map((_, 1))

    val keyByDS: KeyedStream[(String, Int), String] = kvDS.keyBy(_._1)

    /**
      * process： flink底层api 可以操作flink的事件，时间，状态
      *
      */
    val countDS: DataStream[(String, Int)] = keyByDS.process(new MyProcess)

    countDS.print()


    env.execute()
  }

}


class MyProcess extends KeyedProcessFunction[String, (String, Int), (String, Int)] {

  //如果任务中途失败重启，count的值就会丢失
  //var count = 0

  //状态的数据会被checkpoint永久保存起来
  //每一个key都会保存一个状态
  var valueState: ValueState[Int] = _

  //在processElement之前执行，只执行一次
  override def open(parameters: Configuration): Unit = {

    //在这里定义状态


    //获取flink运行环境
    val context: RuntimeContext = getRuntimeContext


    //创建状态描述对象，执行默认值
    val valyeDesc = new ValueStateDescriptor[Int]("count", classOf[Int])

    //冲上下文中获取一个状态
    valueState = context.getState(valyeDesc)


  }

  /**
    * processElement： 每一条数据执行一次，每一个key单独执行
    *
    * @param value ： 一行数据
    * @param ctx   ： 上下文对象
    * @param out   ： 用于向下游发送数据
    */

  override def processElement(value: (String, Int), ctx: KeyedProcessFunction[String, (String, Int), (String, Int)]#Context, out: Collector[(String, Int)]): Unit = {
    //获取之前的状态
    val last: Int = valueState.value()


    //最新统计结果
    val count: Int = last + value._2


    //更新状态
    valueState.update(count)


    //将数据发送到下游
    out.collect((value._1, count))
  }
}